package com.hmall.zookeeper.utils;

import com.hmall.exception.ServiceUnavailableException;
import com.hmall.zookeeper.config.ZookeeperProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class CuratorUtils {
    private static final int BASE_SLEEP_TIME = 1000;
    private static final int MAX_RETRIES = 3;
    private static CuratorFramework zkClient;
    private static final Set<String> REGISTERED_PATH_SET = ConcurrentHashMap.newKeySet();//已注册的节点路径，做缓存
    private static final Map<String, List<String>> SERVICE_ADDRESS_MAP = new ConcurrentHashMap<>();//缓存服务的地址列表，将服务名与服务地址列表做相对映射


    public static CuratorFramework getZkClient() {
        String address = "192.168.100.131:2181";

        if (zkClient != null) {
            return zkClient;
        }
        ExponentialBackoffRetry retry = new ExponentialBackoffRetry(BASE_SLEEP_TIME, MAX_RETRIES);//重试策略，初始等待时间为1秒，最大重试次数为3次，每次重试时间指数增加
        zkClient = CuratorFrameworkFactory.builder()
                .connectString(address)
                .retryPolicy(retry)//设置重拾策略
                .connectionTimeoutMs(5000)
                .namespace("rpc-framework")//名称空间，会在该客户端操作时自动添加/rpc-framework前缀
                .build();
        zkClient.start();
        try {
            if (!zkClient.blockUntilConnected(30, TimeUnit.SECONDS)) {
                throw new IllegalStateException("Zookeeper connection timeout");
            }
        }catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 重新标记线程为中断状态
            throw new ServiceUnavailableException("Connection interrupted", e); // 抛出自定义异常
        }
        return zkClient;
    }

    public static void createPersistentNode(CuratorFramework zkClient, String path) {

        try {
            if (REGISTERED_PATH_SET.contains(path) || zkClient.checkExists().forPath(path) != null) {
                log.info("The node already exists. The node is:[{}]", path);
            }else {
                log.info("The node was created successfully. The node is:[{}]", path);
                zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
                REGISTERED_PATH_SET.add(path);
            }
        } catch (Exception e) {
            log.error("create persistent node for path [{}] fail", path);
        }
    }

    public static List<String> getChildrenNodes(CuratorFramework zkClient, String rpcServiceName) {
        List<String> result = null;
        if (REGISTERED_PATH_SET.contains(rpcServiceName)) {
            result = SERVICE_ADDRESS_MAP.get(rpcServiceName);
            return result;
        }
        String servicePath = "/" + rpcServiceName;
        try {
            result = zkClient.getChildren().forPath(servicePath);
            SERVICE_ADDRESS_MAP.put(rpcServiceName, result);
            registerWatcher(rpcServiceName, zkClient);
        } catch (Exception e) {
            log.error("get children nodes for path [{}] fail", servicePath);
        }
        return result;
    }

    public static void clearRegistry(CuratorFramework zkClient, InetSocketAddress inetSocketAddress) {
        REGISTERED_PATH_SET.stream().parallel().forEach(p -> {
            try {
                if (p.endsWith(inetSocketAddress.toString())) {
                    zkClient.delete().forPath(p);
                }
            } catch (Exception e) {
                log.error("clear registry for path [{}] fail", p);
            }
        });
        log.info("All registered services on the server are cleared:[{}]", REGISTERED_PATH_SET.toString());
    }

    private static void registerWatcher(String rpcServiceName, CuratorFramework zkClient) throws Exception {
        String servicePath = "/" + rpcServiceName;
        PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, servicePath, true);//监听该路径下的子节点
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {

            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                List<String> serverPath = curatorFramework.getChildren().forPath(servicePath);
                SERVICE_ADDRESS_MAP.put(rpcServiceName, serverPath);
            }
        });
        pathChildrenCache.start();
    }
}
